package cn.com.klun.tjy.other.zkUsed;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;

import static org.apache.zookeeper.ZooDefs.Ids.OPEN_ACL_UNSAFE;

public class Client implements Watcher {
    private Logger logger = LoggerFactory.getLogger(Client.class);

    ZooKeeper zk;
    String hostPort;

    public Client(String hostPort) {
        this.hostPort = hostPort;
    }

    void startZk() throws IOException {
        zk = new ZooKeeper(hostPort,15000,this);
    }

    void stopZK() throws InterruptedException {
        zk.close();
    }

    String queueCommand(String command) throws Exception {
        while (true){
            String name = "";
            try{
                name  = zk.create("/tasks/task-",
                        command.getBytes(),
                        OPEN_ACL_UNSAFE,
                        CreateMode.EPHEMERAL_SEQUENTIAL);
                return name;
            } catch (KeeperException.NodeExistsException e){
                throw new Exception(name+"这个节点已经在运行");
            } catch (KeeperException.ConnectionLossException e){
            }
        }

    }
    @Override
    public void process(WatchedEvent watchedEvent) {
        logger.info("yichang "+watchedEvent);
    }

    /**
     * 客户端等待任务的执行结果
     * @param task
     * @param taskCtx
     */
    void submitTask(String task,TaskObject taskCtx){
        //todo测试完成解开 taskCtx.setTask(task);
        taskCtx.setTask(task);
        zk.create("/tasks/task-",
                task.getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT_SEQUENTIAL,
                createTaskCallback,
                taskCtx);

    }
    AsyncCallback.StringCallback createTaskCallback = new AsyncCallback.StringCallback() {
        @Override
        public void processResult(int i, String s, Object o, String s1) {
            switch (KeeperException.Code.get(i)){
                case CONNECTIONLOSS:
                    // 链接丢失，再次提交任务，注意重新提交的任务 可能会导致任务重复
                    submitTask(((String)o),(TaskObject)o);
                    break;
                case OK:
                    logger.info("my created task name:"+s1);
                    // 为这个任务的znode节点设置一个监视点
                    watchStatus("/status/"+s1.replace("/tasks/",""),o);
                    break;
                default:
                    logger.error("运行 异常");
            }
        }
    };
    // 检查状态节点是否存在，并设置监视点
    ConcurrentHashMap<String,Object> ctxMap = new ConcurrentHashMap<>();

    /**
     * 客户端通过该方法传递上下对象，当收到状态节点的通知时，就可以修改这个表示任务的对象
     * @param path
     * @param ctx
     */
    void watchStatus(String path,Object ctx){
        ctxMap.put(path,ctx);
        zk.exists(path,statusWatcher,existsCallback,ctx);
    }
    Watcher statusWatcher = new Watcher() {
        @Override
        public void process(WatchedEvent watchedEvent) {
            if(watchedEvent.getType() == Event.EventType.NodeChildrenChanged){
                assert watchedEvent.getPath().contains("/status/task-");
                zk.getData(watchedEvent.getPath(),false,getDataCallback,ctxMap.get(watchedEvent.getPath()));
            }
        }
    };
    AsyncCallback.StatCallback existsCallback = new AsyncCallback.StatCallback() {
        @Override
        public void processResult(int i, String s, Object o, Stat stat) {
            switch (KeeperException.Code.get(i)){
                case CONNECTIONLOSS:
                    watchStatus(s,o);
                    break;
                case OK:
                    if(stat != null){
                        // 状态节点信息已经存在，所以获取这个节点的信息
                        zk.getData(s,false,getDataCallback,null);
                    }
                    break;
                case NONODE:
                    break;
                    default:
                        logger.error("Something went wrong when "+"checking if the status node exists"+
                        KeeperException.create(KeeperException.Code.get(i),s));
            }
        }
    };


    void getStatusData(String path,Object ctx){
        ctxMap.put(path,ctx);
        // 第二个参数是监视点 false不设置监视点
        zk.getData(path,false,getDataCallback,ctx);
    }
    AsyncCallback.DataCallback getDataCallback = new AsyncCallback.DataCallback() {
        @Override
        public void processResult(int i, String s, Object o, byte[] bytes, Stat stat) {
            switch (KeeperException.Code.get(i)){
                case CONNECTIONLOSS:
                    // 再一次调用这个 getData
                    getStatusData(s,o);
                    break;
                case OK:
                    // 获取到数据的处理方式
                    // TODO 处理获取到的数据保存或者使用whm
                    // 这里会获取到更改的任务 状态 /status/work_id/stat
                    // 根据这个状态 判断这个任务的执行情况 并对这个任务节点进行操作（增删改查 或者 不改变）
                    break;
                default:
                    logger.error("error",KeeperException.create(KeeperException.Code.get(i)));
            }
        }
    };


    public static void main(String[] args) throws Exception {
        Logger logger = LoggerFactory.getLogger(Client.class);
        Client c  = new Client("192.168.137.12:2181");
        c.startZk();
        String name  = c.queueCommand(1+"");
        logger.info("Create"+name);
        Thread.sleep(99999999);
        c.stopZK();
    }
}
